package com.telecom.data; import org.apache.hadoop.conf.Configuration; import 
org.apache.hadoop.hbase.HBaseConfiguration; import 
org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import 
org.apache.hadoop.hbase.util.Bytes; import org.apache.kafka.clients.consumer.*; 
import java.io.IOException; import java.time.Duration; import java.util.Collections; 
import java.util.Properties; public class CleanDataConsumer {
    public static void main(String[] args) {
        // 配置Kafka消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.145.131:9092"); // 使用实际IP
        props.put("group.id", "telecom-group");
        props.put("key.deserializer", 
"org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", 
"org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest"); // 从最早的消息开始消费
        // 配置HBase
        Configuration hbaseConfig = HBaseConfiguration.create();
        hbaseConfig.set("hbase.zookeeper.quorum", "node1");
        hbaseConfig.set("hbase.zookeeper.property.clientPort", "2181");
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
             Connection connection = 
ConnectionFactory.createConnection(hbaseConfig);
             Table table = 
connection.getTable(TableName.valueOf("telecom_call_records"))) {
            consumer.subscribe(Collections.singletonList("telecom_topic"));
            System.out.println("Kafka消费者已启动，开始消费数据...");
            // 消费数据并写入HBase
            while (true) {
                ConsumerRecords<String, String> records = 
consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String[] fields = record.value().split(",");
                    if (fields.length == 4) { // 验证数据格式是否正确
                        String rowKey = fields[0] + "_" + 
System.currentTimeMillis();
                        Put put = new Put(Bytes.toBytes(rowKey));
                        put.addColumn(Bytes.toBytes("info"), 
Bytes.toBytes("caller"), Bytes.toBytes(fields[0]));
                        put.addColumn(Bytes.toBytes("info"), 
Bytes.toBytes("callee"), Bytes.toBytes(fields[1]));
                        put.addColumn(Bytes.toBytes("info"), 
Bytes.toBytes("call_time"), Bytes.toBytes(fields[2]));
                        put.addColumn(Bytes.toBytes("info"), 
Bytes.toBytes("duration"), Bytes.toBytes(Integer.parseInt(fields[3])));
                        table.put(put);
                        System.out.println("已写入HBase: " + rowKey);
                    } else {
                        System.out.println("跳过格式错误的数据: " + record.value());
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
